package com.ai.wxy.spring.storm.test.bolt;

import lombok.extern.slf4j.Slf4j;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

/**
 * 统计单词数量
 *
 * @author 石头
 * @Date 2019/12/4
 * @Version 1.0
 **/
@Slf4j
public class SumBolt extends BaseRichBolt{
    private Map<String, Long> countMap = null;
    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        countMap = new HashMap<String, Long>(100);
    }

    @Override
    public void execute(Tuple tuple) {
        String word = tuple.getStringByField("word");
        Long count  = countMap.get(word);
        if (count == null){
            count = 0L;
        }
        countMap.put(word,++count);
        log.info("单词统计=> {}:{}",word,count);
    }
    @Override
    public void cleanup() {
        countMap.entrySet().stream().forEach(entry -> log.info("【最后结果】==>{}:{}",entry.getKey(),entry.getValue()));
    }
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
